#pragma once
#include <muduo/protobuf/codec.h>
#include <muduo/protobuf/dispatcher.h>
#include <muduo/net/TcpServer.h>
#include <mqcomm/Helper.hpp>
#include <mqcomm/Log.hpp>
#include <mqcomm/Message.pb.h>
#include <mqcomm/protocol.pb.h>
#include <mqcomm/threadpool.hpp>
#include "consumer.hpp"
#include "virtualhost.hpp"
#include "route.hpp"
#include <iostream>
#include <vector>
#include <string>
#include <unordered_map>
#include <memory>
#include <mutex>
#include <functional>
using namespace std;


namespace rbmq {
    typedef shared_ptr<ProtobufCodec> ProtobufCodecPtr;
    typedef shared_ptr<threadpool> threadpoolPtr;

    typedef shared_ptr<mqNs::openChannelRequest> openChannelRequestPtr;
    typedef shared_ptr<mqNs::closeChannelRequest> closeChannelRequestPtr;
    typedef shared_ptr<mqNs::declareExchangeRequest> declareExchangeRequestPtr;
    typedef shared_ptr<mqNs::deleteExchangeRequest> deleteExchangeRequestPtr;
    typedef shared_ptr<mqNs::declareQueueRequest> declareQueueRequestPtr;
    typedef shared_ptr<mqNs::deleteQueueRequest> deleteQueueRequestPtr;
    typedef shared_ptr<mqNs::QueueBindingRequest> QueueBindingRequestPtr;
    typedef shared_ptr<mqNs::QueueUnBindingRequest> QueueUnBindingRequestPtr;
    typedef shared_ptr<mqNs::basicPublshRequest> basicPublshRequestPtr;
    typedef shared_ptr<mqNs::basicAckRequest> basicAckRequestPtr;
    typedef shared_ptr<mqNs::basicConcelRequest> basicConcelRequestPtr;
    typedef shared_ptr<mqNs::basicConsumerRequest> basicConsumerRequestPtr;
    // typedef shared_ptr<mqNs::basicConsumerResponse> basicConsumerResponsePtr;
    // typedef shared_ptr<mqNs::basicCommonResponse> basicCommonResponsePtr;


    // 信道类
    class channel {
    public :
        typedef shared_ptr<channel> ptr;
        channel(){}
        channel(const VirtualHost::ptr& virtualHost, const consumerManager::ptr& cmp, const ProtobufCodecPtr& codec, \
        const muduo::net::TcpConnectionPtr& connection, const string& id, const threadpoolPtr& pool) 
            :_id(id)
            ,_connection(connection)
            ,_codec(codec)
            ,_cmp(cmp)
            ,_virtualHost(virtualHost)
            ,_threadpool(pool)
        {

        }

        // 交换机的声明
        void declareExchange(const declareExchangeRequestPtr& req) {
            // 向虚拟机里面添加一个交换机
            // 但是protobuf里面的map和C++中的unordered_map不是同一种类型(所以我们先要修改一下我们原来的map类型)
            _virtualHost->declareExchange(req->exchange_name(), req->exchange_type(), req->durable(), req->auto_delete(), req->args());
            // 响应的rid和cid要和请求的rid一致
            // 最后进行统一的响应
            basicResponse(req->rid(), req->cid(), true);
        }

        // 删除交换机
        void deleteExchange(const deleteExchangeRequestPtr& req) {
            // 从虚拟机中删除一个交换机
            bool res = _virtualHost->deleteExchange(req->exchange_name());
            basicResponse(req->rid(), req->cid(), res);
        }

        // 队列的声明
        void declareQueue(const declareQueueRequestPtr& req) {
            // 跟队列相关的有：消息、消息管理、绑定信息，消费者(这些信息已经在消费者管理类中已经整合)
            // 先在虚拟机中声明一个队列
            bool res = _virtualHost->declareQueue(req->queue_name(), req->durable(), req->exclusive(), req->auto_delete(), req->args());
            // 如果声明失败，就直接返回响应
            if (!res) {
                basicResponse(req->rid(), req->cid(), false);
                return;
            }
            // 如果成功，还需要声明一个消费者管理句柄，然后再返回响应
            _cmp->InitQueueConsumer(req->queue_name());
            basicResponse(req->rid(), req->cid(), true);
        }

        // 删除队列
        void deleteQueue(const deleteQueueRequestPtr& req) {
            // 直接将于该队列相关的消费者直接干掉
            _cmp->destroyQueueConsumer(req->queue_name());
            // 再从虚拟机中删除队列，返回响应
            _virtualHost->deleteQueue(req->queue_name());
            basicResponse(req->rid(), req->cid(), true);
        }

        // 队列的绑定
        void queueBind(const QueueBindingRequestPtr& req) {
            // 向虚拟机中添加bind信息
            bool res = _virtualHost->Bind(req->queue_name(), req->exchange_name(), req->binding_key());
            // 返回响应
            basicResponse(req->rid(), req->cid(), res);
        }

        // 队列的解绑
        void queueUnBind(const QueueUnBindingRequestPtr& req) {
            // 从虚拟机中解除绑定
            _virtualHost->unBind(req->queue_name(), req->exchange_name());
            basicResponse(req->rid(), req->cid(), true);
        }

        // 消息的发布
        void basicPublish(const basicPublshRequestPtr& req) {
            // 1、判断交换机是否存在，如果不存在直接返回false
            Exchange::ptr exchange = _virtualHost->selectExchange(req->exchange_name());
            if (exchange.get() == nullptr) {
                basicResponse(req->rid(), req->cid(), false);
                return ;
            }
            // 2、获取该交换机所有绑定信息
            MsQueueBindingMap map = _virtualHost->exchangeBindings(req->exchange_name());


            // 3、使用交换机进行交换路由，判断消息可以发送到交换机绑定的哪个队列中(获取到指定交换机的所有绑定信息、以及获取指定的交换机)
            // 因为req里面的properties可能为空，所以也需要先判断req里面是否有properties
            string routing_key;
            mqNs::BasicProperties* properties = nullptr;
            if (req->has_properties()) {
                routing_key = req->properties().routing_key();
                properties = req->mutable_properties();
            }
            // 遍历获取到的交换机的所有binding_key，看看他们和req的routing_key是否匹配，如果匹配，就可以将消息添加到队列中(添加消息管理)
            // 然后向队列的订阅者推送消息(交给线程池完成————将publish绑定进线程池中)
            // 4、将消息添加到队列中(使用虚拟机添加消息的一个管理)
            // 5、向线程池中添加一个消息处理任务(向指定队列的订阅者去推送消息)
            for (auto& binding : map) {
                // 进行路由匹配
                if (Router::route(exchange->_type, binding.second->_binding_key, routing_key)) {
                    auto task = bind(&channel::publish, this, binding.first);
                    // 使用虚拟机将消息添加到队列中
                    _virtualHost->basicPublish(binding.first, properties, req->body());
                    // 将处理消息的任务绑定到线程池中
                    _threadpool->push(task);

                }
            }
            // 返回响应
            basicResponse(req->rid(), req->cid(), true);
        }

        // 消息的确认
        void basicAck(const basicAckRequestPtr& req) {
            // 这个接口如果被调用，则说明收到了消息的确认请求
            // 所以直接通过_virtualHost确认就行了
            _virtualHost->basicAck(req->queue_name(), req->mesage_id());
            basicResponse(req->rid(), req->cid(), true);
        }

        // 订阅队列消息
        void basicConsumer(const basicConsumerRequestPtr& req) {
            // 1、判断队列是否存在
            bool res = _virtualHost->existsQueue(req->queue_name());
            if (!res) {
                return basicResponse(req->rid(), req->cid(), res);
            }
            // 2、如果队列存在，则需要创建对立的一个消费者(难点：需要引入一个回调函数)
            // 3、创建了消费者之后，当前的channel它就变成了一个消费者，所以需要将创建消费者的返回值赋值给我们内部的成员变量_consumer
            auto callback = bind(&channel::consumerCallback, this, placeholders::_1, placeholders::_2, placeholders::_3);
            _consumer = _cmp->createConsumer(req->queue_name(), req->consumer_tag(), req->auto_ack(), callback);
            basicResponse(req->rid(), req->cid(), true);
        }

        // 取消订阅
        void basicCancel(const basicConcelRequestPtr& req) {
            // 从_cmp中取出消费者和队列的订阅关系即可
            _cmp->removeQueueConsumer(req->queue_name(), req->consumer_tag());
            basicResponse(req->rid(), req->cid(), true);

        }

        ~channel() {
            // 如果信道是一个消费者，则需要移除该消费者的订阅
            if (_consumer.get() != nullptr) {
                _cmp->removeQueueConsumer(_consumer->_qname, _consumer->_tag);
            }
        }

    private :
        // 进行统一的响应
        void basicResponse(const string& rid, const string& cid, bool ok) {
            // 构建统一响应对象，使用_codec进行发送
            mqNs::basicCommonResponse resp;
            resp.set_ok(ok);
            resp.set_rid(rid);
            resp.set_cid(cid);
            _codec->send(_connection, resp);
        }
        // 向指定队列推送消息
        void publish(const string& queue_name) {
            // 指定队列消费消息
            // 1、从队列中取出一条消息
            MessagePtr msg = _virtualHost->basicConsume(queue_name);
            if (msg.get() == nullptr) {
                LOG(Debug) << "向指定队列: " << queue_name << " 推送消息失败: 该队列中没有消息!" << endl;
                return;
            }

            // 2、从队列订阅者中取出一个订阅者(也就是消费者)
            consumer::ptr consumerp = _cmp->choose(queue_name);
            if (consumerp.get() == nullptr) {
                LOG(Debug) << "向指定队列: " << queue_name << " 推送消息失败: 该队列没有消费者订阅!" << endl;
                return;                
            }

            // 3、调用订阅者的消息处理回调函数，实现消息的推送
            consumerp->_callback(consumerp->_tag, msg->mutable_payload()->mutable_properties(), msg->payload().body());

            // 4、判断：如果订阅者是一个自动确认，就不需要等待确认，直接删除消息，否则就需要等到外部收到消息确认后再删除
            if (consumerp->_auto_ack == true) {
                _virtualHost->basicAck(queue_name, msg->payload().properties().id());
            }
        }

        // 消费者的回调函数
        void consumerCallback(const string& tag, const mqNs::BasicProperties* bp, const string& msg_body) {
            // 针对参数，组织出一个消息推送响应，将消息推送给对应的客户端
            mqNs::basicConsumerResponse resp;
            resp.set_cid(_id);
            resp.set_body(msg_body);
            resp.set_consumer_tag(tag);
            if (bp) {
                resp.mutable_properties()->set_id(bp->id());
                resp.mutable_properties()->set_delivery_mode(bp->delivery_mode());
                resp.mutable_properties()->set_routing_key(bp->routing_key());
            }
            
            // 通过_codec发送
            _codec->send(_connection, resp);
        }
        
    private :
        string _id; // 信道的id
        consumer::ptr _consumer; // 信道所关联的消费者
        muduo::net::TcpConnectionPtr _connection; // 信道所关联的连接
        ProtobufCodecPtr _codec; // 信道的protobuf协议处理器
        consumerManager::ptr _cmp; // 消费者管理句柄
        VirtualHost::ptr _virtualHost; // 虚拟机管理句柄
        threadpoolPtr _threadpool; // 线程池
    };
    // ===========================================================================================================================================

    // 信道的管理类
    class channelManager {
        public :
            typedef shared_ptr<channelManager> ptr;
            channelManager() {

            }

            // 打开一个信道
            bool openChannel(const VirtualHost::ptr& virtualHost, const consumerManager::ptr& cmp, const ProtobufCodecPtr& codec, \
                            const muduo::net::TcpConnectionPtr& connection, const string& id, const threadpoolPtr& pool) 
            {
                unique_lock<mutex> lock(_mutex);
                // 先判断Channel是否已经存在，如果已经存在就不能再打开
                auto it = _channels.find(id);
                if (it != _channels.end()) {
                    return false;
                }
                channel::ptr ch = make_shared<channel>(virtualHost, cmp, codec, connection, id, pool);
                _channels.insert(make_pair(id, ch));
                return true;
            }
            
            // 关闭一个信道
            void closeChannel(const string& id) {
                unique_lock<mutex> lock(_mutex);
                _channels.erase(id);
            }

            // 获取指定信道
            channel::ptr getChannel(const string& id) {
                unique_lock<mutex> lock(_mutex);
                auto it = _channels.find(id);
                if (it == _channels.end()) {
                    return channel::ptr();
                }
                return it->second;
            }

        private :
            mutex _mutex; // 互斥锁
            unordered_map<string, channel::ptr> _channels; // 信道id与信道管理句柄的映射哈希表
    };
};
